package com.bigdata.spark.mallapp_realtime.test

import java.lang
import java.text.SimpleDateFormat
import java.util.Date

import com.alibaba.fastjson.{JSON, JSONObject}
import com.bigdata.spark.util.{MyKafkaUtil, OffsetManagerUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object mall_log_Exactly_once_action_1 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("action1")
    val ssc: StreamingContext = new StreamingContext(conf,Seconds(5))
    var topic:String = "gmall_event_0523"
    var groupId = "action1"

    //从Redis中获取Kafka分区偏移量。Redis中的key是topic+groupId，值是map类型的TopicPartition+offset
    val offsetMap: Map[TopicPartition, Long] = OffsetManagerUtil.getOffset(topic,groupId)


    var recordDStream: InputDStream[ConsumerRecord[String, String]] = null
    if(offsetMap!=null && offsetMap.size >0){
      //如果Redis中存在当前消费者组对该主题的偏移量信息，那么从执行的偏移量位置开始消费
      recordDStream = MyKafkaUtil.getKafkaStream(topic,ssc,offsetMap,groupId)
    }else{
      //如果Redis中没有当前消费者组对该主题的偏移量信息，那么还是按照配置，从最新位置开始消费
      recordDStream = MyKafkaUtil.getKafkaStream(topic,ssc,groupId)
    }

    //获取当前采集周期从Kafka中消费的数据的起始偏移量以及结束偏移量值,目的是，逻辑完毕之后更新偏移量。获取结束偏移量，要从
    //现在读取到的recordDStream这个离散化流获取。这个用getKafkaStream创建的离散化流底层是kafkaRDD，这才可以取到起始结束偏移量值。

    //创建读取数据用的array
    var offsetRanges: Array[OffsetRange] = Array.empty[OffsetRange]  //就是new Array[offsetRange](0)


    //可以实现只取值，不提交改变的算子：transform。
    val offsetDStream: DStream[ConsumerRecord[String, String]] = recordDStream.transform {
      rdd => {

        //取到起始结束偏移量
        //因为recodeDStream底层封装的是KafkaRDD，混入了HasOffsetRanges特质，这个特质中提供了可以获取偏移量范围的方法
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges  //这个属性是private的，不能随便访问
        //RDD里没有offsetRanges方法。要访问它的父类来。所以要把RDD转换成HasOffsetRanges。RDD强制类型转换为父类HasOffsetRanges。
        //tips：存在继承关系才能强制转换。kafkaRDD子，HasOffsetRanges是父。（所以叫做KafkaRDD混入了HasOffsetRanges特质）
        rdd
      }
    }
    //offsetRanges这样就取得了，一次获取的是所有分区内所有partition的分区，主题，起始结束

    //   其中单条offsetRange: 数据，获得了读取kafka时的分区，主题，起始结束。
    //    val topic: String,
    //    val partition: Int,
    //    val fromOffset: Long,
    //    val untilOffset: Long) extends Serializable


    val Date: DStream[(String, lang.Long)] = offsetDStream.map { ConsumerRecord =>
      val jsonStr: String = ConsumerRecord.value()

      val jsonObj: JSONObject = JSON.parseObject(jsonStr)
      //获取时间戳 毫秒数
      val ts: lang.Long = jsonObj.getLong("ts")

      //      //获取字符串 日期 小时

      val dateHourString: String = new SimpleDateFormat("yyyy-MM-dd HH").format(new Date(ts))

      (dateHourString, ts)
    }





    //保存的时候需要幂等性。

    Date.foreachRDD(  //foreachRDD的用法：里面的参数是个迭代器。
      // 这是保存数据用的算子。对本时间间隔，执行一次。这里是5秒。在一个为批次的处理时间间隔里， DStream只产生一个RDD
      rdd=>{


        rdd.foreach(
          record=>{
            println(record)
          }

        )

            println("第一层")
            println(rdd,rdd.getClass)
            //提交偏移量到Redis中，以分区为单位。
            //println("提交偏移量到Redis中，以分区为单位。")
            OffsetManagerUtil.saveOffset(topic,groupId,offsetRanges)

      }


    )





    ssc.start()
    ssc.awaitTermination()


  }
}
